/**
 * Software License, Version 1.0
 *
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 *
 *
 *Redistribution and use in source and binary forms, with or without
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For
 * further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and
 * wherever such third-party acknowledgments normally appear.
 *
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering,
 * shall not be used to endorse or promote products derived from this software
 * without prior written permission from Indiana University.  For written
 * permission, please contact the Advanced Research and Technology Institute
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering,
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 *
 *
 * Indiana University provides no reassurances that the source code provided
 * does not infringe the patent or any other intellectual property rights of
 * any other entity.  Indiana University disclaims any liability to any
 * recipient for claims brought by any other entity based on infringement of
 * intellectual property rights or otherwise.
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.service.replay.impl;

import java.util.Arrays;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.LinkedList;
import java.util.Vector;

import cgl.narada.event.NBEvent;
import cgl.narada.event.NBEventTypes;
import cgl.narada.event.TemplateProfileAndSynopsisTypes;
import cgl.narada.event.impl.NBEventGenerator;
import cgl.narada.matching.CrossMatching;
import cgl.narada.matching.Profile;
import cgl.narada.service.ServiceException;
import cgl.narada.service.ServiceExceptionListener;
import cgl.narada.service.client.ClientService;
import cgl.narada.service.client.EventConsumer;
import cgl.narada.service.client.EventProducer;
import cgl.narada.service.client.NBEventListener;
import cgl.narada.service.client.SessionService;
import cgl.narada.service.replay.ReplayEvent;
import cgl.narada.service.replay.ReplayRequest;
import cgl.narada.service.replay.ReplayResponse;
import cgl.narada.service.replay.ReplayService;
import cgl.narada.service.replay.ReplayServiceExchange;
import cgl.narada.service.storage.InventoryEvent;
import cgl.narada.service.storage.StorageService;

/**
 The Replay Service interface that is used by the consumers to play back
 events that were previously archived. Replays are initiated by
 ReplayRequests.

 @author Shrideep Pallickara
 $Date$
 $Revision$
 */

public class ReplayServiceImpl extends ReplayService implements
        NBEventListener, ServiceExceptionListener {

    private int MAX_SEQUENCE_RANGE_IN_REQUEST = 1000;
    private int MAX_NUM_OF_PROFILES_IN_REQUEST = 20;
    private int MAX_NUM_OF_SEQUENCES_IN_REPLAY = 1000;
    private int MAX_PENDING_REQUESTS = 50;

    private StorageService storageService;
    private int replayEntityId;

    private ClientService clientService;
    private EventConsumer eventConsumer;
    private EventProducer eventProducer;

    private CrossMatching crossMatching;

    private Hashtable replayServiceProfiles;
    private String replayTopicString = "/ReplayServices";

    /** Maintains a list of templates and the list of entities registered to this
     template. template-id's to a Hashtable of registered entities */
    private Hashtable templatesAndEntities;
    private boolean initialized = false;
    private static ReplayService replayService = new ReplayServiceImpl();
    private String moduleName = "ReplayServiceImpl: ";

    private long[] sequences;
    private long[] timestamps;
    private long sequence;
    private long timestamp;

    private ReplayServiceImpl() {
        replayServiceProfiles = new Hashtable();
        crossMatching = new CrossMatching();
    }

    public void initialize(int replayEntityId, StorageService storageService,
                           Hashtable templatesAndEntities) throws
            ServiceException {
        if (initialized) {
            return;
        }
        this.replayEntityId = replayEntityId;
        this.storageService = storageService;
        this.templatesAndEntities = templatesAndEntities;
        initializeProducerConsumer();
        initialized = true;
    }

    public boolean isInitialized() {
        return initialized;
    }

    public static ReplayService getInstance() {
        return replayService;
    }

    private void initializeProducerConsumer() throws ServiceException {
        clientService = SessionService.getClientService(replayEntityId);
        if (!clientService.isInitialized()) {
            throw new ServiceException(moduleName +
                                       "Client service for entity="
                                       + replayEntityId +
                    " should have been previously initialized");
        }
        eventConsumer = clientService.createEventConsumer(this);
        eventProducer = clientService.createEventProducer();
        Enumeration e = templatesAndEntities.keys();
        while (e.hasMoreElements()) {
            int templateId = ((Integer) e.nextElement()).intValue();
            registerSubscriptionsForTemplate(templateId);
        }
    }

    /** This method registers subscriptions for templates */
    public void registerSubscriptionsForTemplate(int templateId) throws
            ServiceException {
        Object templateKey = new Integer(templateId);
        if (replayServiceProfiles.containsKey(templateKey)) {
            /**
             * We have already performed the necessary suscriptions for this
             * template in question
             */
            return;
        }
        String subscriptionString = replayTopicString + "/" + templateId;
        Profile profile = clientService.createProfile(
                TemplateProfileAndSynopsisTypes.STRING, subscriptionString);
        eventConsumer.subscribeTo(profile);
        replayServiceProfiles.put(templateKey, profile);
    }

    /** This method de-registers subscriptions for a template */
    public void deregisterSubscriptionsForTemplate(int templateId) throws
            ServiceException {
        Object templateKey = new Integer(templateId);
        if (!replayServiceProfiles.containsKey(templateKey)) {
            /**
             * We have already performed the necessary suscriptions for this
             * template in question
             */
            throw new ServiceException(moduleName
                                       +
                    "Unknown template submitted for removal ");
        }
        Profile profileToUnsubscribe = (Profile) replayServiceProfiles
                                       .get(templateKey);
        eventConsumer.unSubscribe(profileToUnsubscribe);
    }

    /** Process a replay request */
    public void processReplayRequest(ReplayRequest replayRequest) {
        int templateId = replayRequest.getTemplateId();
        int entityId = replayRequest.getEntityId();
        String reasonForFailure = null;
        /** Check authorizations */
        reasonForFailure = processTemplateAndEntity(templateId, entityId);
        if (reasonForFailure != null) {
            createAndIssueReplayResponse(replayRequest, false, reasonForFailure, null);
            return;
        }
        long[] replaySequences = null;
        if (replayRequest.hasRequestForTimestamp()) {
            long seqNum = replayRequest.getSequenceNumber();
            if (seqNum < 0) {
                LinkedList result = null;
                if (seqNum == -1) { // minimum case
                    result = this.getMinimumSequenceNumberAndTimestamp(
                            replayRequest
                            .getTemplateId());
                } else if (seqNum == -2) { // maximum case
                    result = this.getMaximumSequenceNumberAndTimestamp(
                            replayRequest
                            .getTemplateId());
                }
                if (result == null) {
                    createAndIssueReplayResponse(replayRequest, false,
                                                 "no minimum/maximum sequence number and timestamp", null);
                } else if ((result.getFirst() == null) || (result.getLast() == null)) {
                    createAndIssueReplayResponse(replayRequest, false,
                                                 "no minimum/maximum sequence number and timestamp", null);
                } else {
                    createAndIssueReplayResponse(replayRequest, true, null,
                                                 Long
                                                 .parseLong(result.getFirst().
                            toString()), Long.parseLong(result
                            .getLast().toString()));
                }
                return;
            }
            return;
        }
        if (replayRequest.hasRequestSequencesAndTimestamps()) {
            reasonForFailure = getSequencesAndTimestamps(templateId);
            if (sequences == null) {
                System.out.println("Sequences are null");
            }
            if (reasonForFailure != null) {
                createAndIssueReplayResponse(replayRequest, false,
                                             reasonForFailure,
                                             null);
            } else {
                createAndIssueReplayResponse(replayRequest, true, null,
                                             sequences,
                                             timestamps);
            }
            return;
        }
        if (replayRequest.hasListOfSequencesToReplay()) {
            replaySequences = replayRequest.getListOfSequencesToReplay();
        } else if (replayRequest.hasSpecifiedSequenceRange()) {
            reasonForFailure = processValidityOfSequenceRange(replayRequest);
            if (reasonForFailure != null) {
                createAndIssueReplayResponse(replayRequest, false,
                                             reasonForFailure,
                                             null);
                return;
            }
            long sequenceRangeStart = replayRequest.getStartingSequenceNumber();
            long sequenceRangeEnd = replayRequest.getEndingSequenceNumber();
            if (sequenceRangeEnd == 0) {
                long maxSequence = 0;
                try {
                    maxSequence = storageService.getSequenceNumberLastAssigned();
                } catch (ServiceException serEx) {
                    reasonForFailure = serEx.toString();
                    createAndIssueReplayResponse(replayRequest, false,
                                                 reasonForFailure,
                                                 null);
                    return;
                }
                if (sequenceRangeStart +
                    MAX_SEQUENCE_RANGE_IN_REQUEST < maxSequence) {
                    sequenceRangeEnd = sequenceRangeStart +
                                       MAX_SEQUENCE_RANGE_IN_REQUEST;
                } else {
                    sequenceRangeEnd = maxSequence;
                }
            }
            try {
                replaySequences = storageService.retrieveSequences(
                        sequenceRangeStart,
                        sequenceRangeEnd, templateId,
                        MAX_NUM_OF_SEQUENCES_IN_REPLAY);
            } catch (ServiceException serEx) {
                reasonForFailure = serEx.toString();
                createAndIssueReplayResponse(replayRequest, false,
                                             reasonForFailure,
                                             null);
                return;
            }
            if (replaySequences == null) {
                reasonForFailure =
                        "No events to replay between sequence range ("
                        + sequenceRangeStart + ") <--> (" + sequenceRangeEnd
                        + ") for template [" + templateId + "]";
                createAndIssueReplayResponse(replayRequest, false,
                                             reasonForFailure,
                                             null);
                return;
            }
        } else if (replayRequest.hasReplayWithTimestampRange()) {
            reasonForFailure = processValidityOfSequenceRange(replayRequest);
            if (reasonForFailure != null) {
                createAndIssueReplayResponse(replayRequest, false,
                                             reasonForFailure,
                                             null);
                return;
            }
            reasonForFailure = this.getSequencesAndTimestamps(replayRequest
                    .getStartTimestamp(), replayRequest.getEndTimestamp(),
                    templateId);
            if (reasonForFailure != null) {
                createAndIssueReplayResponse(replayRequest, false,
                                             reasonForFailure,
                                             null);
                return;
            }
            replaySequences = sequences;
            if (replaySequences == null) {
                reasonForFailure =
                        "No events to replay between timestamp range ("
                        + replayRequest.getStartTimestamp() + ") <--> ("
                        + replayRequest.getEndTimestamp() + ") for template [" +
                        templateId
                        + "]";
                createAndIssueReplayResponse(replayRequest, false,
                                             reasonForFailure,
                                             null);
                return;
            }
            createAndIssueReplayResponse(replayRequest, true, null,
                                         replaySequences, sequences,
                                         timestamps);
            return;
        }

        if (replayRequest.hasSpecifiedConstraints()) {
            processSequencesWithSpecifiedConstraints(replaySequences,
                    replayRequest);
        } else {
            createAndIssueReplayResponse(replayRequest, true, null,
                                         replaySequences);
        }
        return;
    }
    /*end process Replay Request */

    private String getTimestamp(int templateId, long sequence, long timestamp) {
        String reasonForFailure = null;
        long result = 0;

        try {
            result = storageService.retrieveTimestamp(templateId, sequence);
        } catch (ServiceException serEx) {
            reasonForFailure = serEx.toString();
            return reasonForFailure;
        }
        timestamp = result;
        return null;
    }

    private String getSequencesAndTimestamps(int templateId) {
        String reasonForFailure = null;
        LinkedList result = null;
        try {
            result = storageService.retrieveSequencesAndTimestamps(templateId);
        } catch (ServiceException serEx) {
            reasonForFailure = serEx.toString();
            return reasonForFailure;
        }
        int listSize = result.size();
        int length = listSize / 2;
        this.sequences = new long[length];
        this.timestamps = new long[length];
        int j = 0;
        for (int i = 0; i < length; i++) {
            sequences[i] = Long.parseLong(result.get(j).toString());
            timestamps[i] = Long.parseLong(result.get(j + 1).toString());
            j = j + 2;
        }
        return reasonForFailure;
    }

    private String getSequencesAndTimestamps(long startingTimestamp,
                                             long endingTimestamp,
                                             int templateId) {
        String reasonForFailure = null;
        LinkedList result = null;
        try {
            result = storageService.retrieveSequencesAndTimestamps(
                    startingTimestamp,
                    endingTimestamp, templateId);
        } catch (ServiceException serEx) {
            reasonForFailure = serEx.toString();
            return reasonForFailure;
        }
        if (result == null) {
            reasonForFailure =
                    "Result set is null. No Sequences found in the range";
            return reasonForFailure;
        }
        int listSize = result.size();
        int length = listSize / 2;
        this.sequences = new long[length];
        this.timestamps = new long[length];
        int j = 0;
        for (int i = 0; i < length; i++) {
            sequences[i] = Long.parseLong(result.get(j).toString());
            timestamps[i] = Long.parseLong(result.get(j + 1).toString());
            j = j + 2;
        }
        return reasonForFailure;
    }

    private String processTemplateAndEntity(int templateId, int entityId) {
        String reasonForFailure = null;
        Object templateKey = new Integer(templateId);
        Object entityKey = new Integer(entityId);
        if (!templatesAndEntities.containsKey(templateKey)) {
            reasonForFailure =
                    "The replay service does not know about the template"
                    + " in question";
            return reasonForFailure;
        }
        Hashtable entityList = (Hashtable) templatesAndEntities.get(templateKey);
        if (!entityList.containsKey(entityKey)) {
            reasonForFailure = "Entity [" + entityId +
                               "] is not authorized to "
                               + "initiate replays on template identifier (" +
                               templateId + ")";
            return reasonForFailure;
        }
        return reasonForFailure;
    }

    private String processValidityOfReplaySequences(int templateId,
            long[] replaySequences) {
        String reasonForFailure = null;
        if (replaySequences.length > MAX_NUM_OF_SEQUENCES_IN_REPLAY) {
            reasonForFailure = "A total of (" + replaySequences.length
                               +
                    ") sequences were requested for replay. This is greater than the "
                               + " maximum allowed number of [" +
                               MAX_NUM_OF_SEQUENCES_IN_REPLAY
                               +
                    "] sequences that are allowed in a replay request";
            return reasonForFailure;
        }
        /* if (>MAX_NUM_OF_SEQUENCES_IN_REPLAY)*/
        if (replaySequences.length == 0) {
            reasonForFailure = "A total of (" + replaySequences.length
                               +
                    ") sequences were requested for replay. This is not valid";
            return reasonForFailure;
        }
        /* if (sequenceLength == 0)*/
        Arrays.sort(replaySequences);
        int index = replaySequences.length - 1;
        long maxSequenceNum = 0;
        try {
            maxSequenceNum = storageService.getSequenceNumberLastAssigned();
        } catch (ServiceException serEx) {
            reasonForFailure = serEx.toString();
            return reasonForFailure;
        }
        if (replaySequences[index] > maxSequenceNum) {
            reasonForFailure = "The sequence number (" + replaySequences[index]
                               +
                    ") specified for replay is greater than the maximum sequence ("
                               + maxSequenceNum +
                               ") currently available for replay";
            return reasonForFailure;
        }
        /** check if the replaySequences listed are for the specific templateId
         that is in question. */
        long[] incorrectSequences = null;
        try {
            incorrectSequences = storageService.
                                 checkIfSequencesBelongToTemplate(
                                         templateId, replaySequences);
        } catch (ServiceException serEx) {
            onServiceException(serEx);
            reasonForFailure = serEx.toString();
            return reasonForFailure;
        }
        if (incorrectSequences != null) {
            reasonForFailure =
                    "Some of the specified sequences DO NOT belong to "
                    + "the specified template=" + templateId
                    + ". Offending sequences are ->";
            for (int i = 0; i < incorrectSequences.length; i++) {
                reasonForFailure += ("[" + incorrectSequences[i] + "] ");
            }
            return reasonForFailure;
        }
        return reasonForFailure;
    }

    private String processValidityOfSequenceRange(ReplayRequest replayRequest) {
        String reasonForFailure = null;
        long sequenceRangeStart = replayRequest.getStartingSequenceNumber();
        long sequenceRangeEnd = replayRequest.getEndingSequenceNumber();
        if (sequenceRangeStart > sequenceRangeEnd) {
            reasonForFailure = "The start=(" + sequenceRangeStart + ") of the "
                               +
                    "specified sequence range is GREATER THAN the end=("
                               + sequenceRangeEnd +
                               ") contained within the replay request.";
            return reasonForFailure;
        }
        long sequenceRange = sequenceRangeEnd - sequenceRangeStart;
        if (sequenceRange > MAX_SEQUENCE_RANGE_IN_REQUEST) {
            reasonForFailure = "The sequence range of ("
                               + sequenceRange
                               +
                    ") specified within the request is GREATER than the MAXIMUM ALLOWED "
                               + "RANGE = " + MAX_SEQUENCE_RANGE_IN_REQUEST +
                               " specified by the "
                               + "REPLAY service.";
            return reasonForFailure;
        }
        long maxSequenceNum = 0;
        try {
            maxSequenceNum = storageService.getSequenceNumberLastAssigned();
        } catch (ServiceException serEx) {
            reasonForFailure = serEx.toString();
            return reasonForFailure;
        }
        long offendingSequence = 0;
        if (sequenceRangeEnd > maxSequenceNum) {
            offendingSequence = sequenceRangeEnd;
        }
        if (sequenceRangeStart > maxSequenceNum) {
            offendingSequence = sequenceRangeStart;
        }
        if (offendingSequence != 0) {
            reasonForFailure = "The sequence number (" + offendingSequence
                               +
                    ") specified for replay is greater than the maximum sequence ("
                               + maxSequenceNum +
                               ") currently available for replay";
            return reasonForFailure;
        }
        return reasonForFailure;
    }

    private void processSequencesWithSpecifiedConstraints(long[]
            replaySequences,
            ReplayRequest replayRequest) {
        String reasonForFailure = null;
        String additionalInfo = null;
        if (replayRequest.hasSpecifiedConstraints()) {
            Profile[] constraints = replayRequest.getListOfSpecifiedConstraints();
            long[] matchingSequences = computeSubSequencesMatchingConstraints(
                    replaySequences, constraints);
            if (matchingSequences == null) {
                reasonForFailure = "There were ("
                                   + replaySequences.length
                                   + ") that"
                                   +
                        " were considered for replay. Of this none of corresponding events"
                                   +
                        " satisfied any of the specified constraints";
                createAndIssueReplayResponse(replayRequest, false,
                                             reasonForFailure,
                                             null);
                return;
            }
            additionalInfo = "There were (" + replaySequences.length
                             + ") that were considered for replay. Of this ("
                             + matchingSequences.length +
                             ") satisfied the specified constraints";
            createAndIssueReplayResponse(replayRequest, true, additionalInfo,
                                         matchingSequences);
            return;
        }
        additionalInfo = "There were ("
                         + replaySequences.length
                         +
                ") that were considered for replay. Since no additional constraints "
                         +
                " have been specified, the replay service will route all the sequences";
        createAndIssueReplayResponse(replayRequest, true, additionalInfo,
                                     replaySequences);
    }

    private long[] computeSubSequencesMatchingConstraints(long[]
            replaySequences,
            Profile[] constraints) {
        Vector matchingSequences = new Vector();
        try {
            for (int i = 0; i < replaySequences.length; i++) {
                InventoryEvent inventoryEvent = storageService
                                                .getStoredEvent(replaySequences[
                        i]);
                NBEvent nbEvent = inventoryEvent.getEvent();
                boolean eventMatchesConstraints = crossMatching
                                                  .
                        checkIfEventMatchesConstraints(nbEvent, constraints);
                if (eventMatchesConstraints) {
                    matchingSequences.addElement(new Long(replaySequences[i]));
                }
            }

        } catch (ServiceException serEx) {
            onServiceException(serEx);
        }
        int subSequenceLength = matchingSequences.size();
        if (subSequenceLength == 0) {
            return null;
        }
        long[] subSequence = new long[subSequenceLength];
        for (int i = 0; i < subSequenceLength; i++) {
            subSequence[i] = ((Long) matchingSequences.elementAt(i)).longValue();
        }
        return subSequence;
    }

    private void createAndIssueReplayResponse(ReplayRequest replayRequest,
                                              boolean success,
                                              String additionalInfo,
                                              long[] sequencesToBeReplayed) {

        int templateId = replayRequest.getTemplateId();
        int entityId = replayRequest.getEntityId();
        String replayIdentifier = replayRequest.getReplayIdentifier();

        ReplayResponse replayResponse = null;
        if (replayRequest.getReplayTopic() != null) {
            replayResponse = new ReplayResponseEvent(replayRequest, success,
                    additionalInfo, sequencesToBeReplayed);
        } else {
            replayResponse = new ReplayResponseEvent(templateId, entityId,
                    replayIdentifier, success, additionalInfo,
                    sequencesToBeReplayed);
        }

        try {
            issueReplayResponse(replayResponse);
            if (sequencesToBeReplayed != null) {
                issueReplayEvents(replayResponse, sequencesToBeReplayed);
            } else {
                System.out.println("no sequences to be played");
            }
        } catch (ServiceException serEx) {
            onServiceException(serEx);
        }

    }

    private void createAndIssueReplayResponse(ReplayRequest replayRequest,
                                              boolean success,
                                              String additionalInfo,
                                              long[] sequencesToBeReplayed,
                                              long[] sequences,
                                              long[] timestamps) {

        int templateId = replayRequest.getTemplateId();
        int entityId = replayRequest.getEntityId();
        String replayIdentifier = replayRequest.getReplayIdentifier();

        ReplayResponse replayResponse = null;
        if ((sequences != null) && (timestamps != null)) {
//			System.out.println("sequence and timestamp not null");
            replayResponse = new ReplayResponseEvent(replayRequest, success,
                    additionalInfo, true, sequences, timestamps);
        }
        if (replayRequest.getReplayTopic() == null) {
            replayResponse = new ReplayResponseEvent(templateId, entityId,
                    replayIdentifier, success, additionalInfo,
                    sequencesToBeReplayed);
        }

        try {
            issueReplayResponse(replayResponse);
            if (sequencesToBeReplayed != null) {
                issueReplayEvents(replayResponse, sequencesToBeReplayed);
            } else {
                System.out.println("no sequences to be played");
            }
        } catch (ServiceException serEx) {
            onServiceException(serEx);
        }

    }

    private void createAndIssueReplayResponse(ReplayRequest replayRequest,
                                              boolean success,
                                              String additionalInfo,
                                              long[] sequences,
                                              long[] timestamps) {

        ReplayResponse replayResponse = null;
        if ((sequences != null) && (timestamps != null)) {
            replayResponse = new ReplayResponseEvent(replayRequest, success,
                    additionalInfo, true, sequences, timestamps);
        } else {
            System.out.println("sequence and timestamp null");
            replayResponse = new ReplayResponseEvent(replayRequest, success,
                    additionalInfo, false, null, null);
        }

        try {
            issueReplayResponse(replayResponse);
        } catch (ServiceException serEx) {
            onServiceException(serEx);
        }

    }

    private void createAndIssueReplayResponse(ReplayRequest replayRequest,
                                              boolean success,
                                              String additionalInfo,
                                              long sequence, long timestamp) {

        ReplayResponse replayResponse = null;
        if (timestamp != 0) {
            replayResponse = new ReplayResponseEvent(replayRequest, success,
                    additionalInfo, true, sequence, timestamp);
        } else {
            replayResponse = new ReplayResponseEvent(replayRequest, success,
                    additionalInfo, false, 0, 0);
        }

        try {
            issueReplayResponse(replayResponse);
        } catch (ServiceException serEx) {
            onServiceException(serEx);
        }

    }

    private void issueReplayResponse(ReplayResponse replayResponse) throws
            ServiceException {
        System.out.println(moduleName + "Issuing replay response ...");
        int entityId = replayResponse.getEntityId();

        NBEvent nbEvent = eventProducer.generateEvent(
                TemplateProfileAndSynopsisTypes.STRING,
                entityId + replayTopicString,
                replayResponse.getBytes());
        NBEventGenerator.setEventType(nbEvent, NBEventTypes.REPLAY_EXCHANGE);
        eventProducer.publishEvent(nbEvent);
    }

    private void issueReplayEvents(ReplayResponse replayResponse,
                                   long[] sequencesToBeReplayed) throws
            ServiceException {
        System.out.println(moduleName + "Issuing replay events ...");
        int entityId = replayResponse.getEntityId();
        String replayIdentifier = replayResponse.getReplayIdentifier();
        int numOfSequences = sequencesToBeReplayed.length;
        String replayTopic = entityId + replayTopicString;
        if (replayResponse.getReplayTopic() != null) {
            replayTopic = replayResponse.getReplayTopic();
        }
        System.out.println("Topic: " + replayTopic);
        for (int i = 0; i < numOfSequences; i++) {
            long sequenceToGo = sequencesToBeReplayed[i];
            /** Proceed to retrive inventory event and create the replay event */
            InventoryEvent inventoryEvent = storageService
                                            .getStoredEvent(sequenceToGo);
            if (inventoryEvent == null) {
                System.out.println("NULL INVENTORY EVENT:" + sequenceToGo);
                continue;
            }
            NBEvent nbEvent = inventoryEvent.getEvent();
            //      System.out.println("*******");
            //      System.out.println("NBEvent topic:"+nbEvent.getContentSynopsis().toString());
            //      System.out.println("*******");
            ReplayEvent replayEvent = new ReplayEventImpl(sequenceToGo, nbEvent,
                    replayIdentifier);
            /** Proceed to publish the replay event */
            NBEvent nbReplayEvent = eventProducer.generateEvent(
                    TemplateProfileAndSynopsisTypes.STRING, replayTopic,
                    replayEvent
                    .getBytes());
            NBEventGenerator
                    .setEventType(nbReplayEvent, NBEventTypes.REPLAY_EXCHANGE);
            eventProducer.publishEvent(nbReplayEvent);
        }
    }

    /** Set the upper bound for the maximum range in replay sequences */
    public void setSequenceRangeInReplayRequest(int maxSequenceRange) {
        MAX_SEQUENCE_RANGE_IN_REQUEST = maxSequenceRange;
    }

    /** Set the maximum number of profile constraints within a replay request */
    public void setMaximumNumberOfProfilesInReplayRequest(int maxConstraints) {
        MAX_NUM_OF_PROFILES_IN_REQUEST = maxConstraints;
    }

    /** Set Maximum number of sequences in a replay */
    public void setMaximumNumberOfSequencesInReplay(int maxNumOfSequences) {
        MAX_NUM_OF_SEQUENCES_IN_REPLAY = maxNumOfSequences;
    }

    /** Set a limit on the maximum number of pending requests */
    public void setMaximumNumberOfPendingRequests(int maxPendingRequests) {
        MAX_PENDING_REQUESTS = maxPendingRequests;
    }

    public long getStoredEventTimestamp(long seqNumber) {
        long ts = 0;
        try {
            ts = storageService.getStoredEvent(seqNumber).getEvent()
                 .getEventHeaders().getTimeStamp();
        } catch (ServiceException e) {
            ts = 0;
        }
        return ts;
    }

    public LinkedList getMinimumSequenceNumberAndTimestamp(int templateId) {
        LinkedList result = null;
        try {
            result = storageService
                     .retrieveMinimumSequeceNumberAndTimestamp(templateId);
        } catch (ServiceException e) {
            e.printStackTrace();
            result = null;
        }
        return result;
    }

    public LinkedList getMaximumSequenceNumberAndTimestamp(int templateId) {
        LinkedList result = null;
        try {
            result = storageService
                     .retrieveMaximumSequeceNumberAndTimestamp(templateId);
        } catch (ServiceException e) {
            e.printStackTrace();
            result = null;
        }
        return result;
    }

    public void onEvent(NBEvent nbEvent) {
        int eventType = nbEvent.getEventType();
        byte[] contentPayload = nbEvent.getContentPayload();
        byte exchangeType = contentPayload[0];
        if (eventType == NBEventTypes.NORMAL_EVENT) {
            System.out.println(moduleName + "Received NORMAL event \n\n");
            return;
        }
        if (eventType == NBEventTypes.REPLAY_EXCHANGE) {
            if (exchangeType == ReplayServiceExchange.REPLAY_SERVICE_REQUEST) {
                ReplayRequest replayRequest = new ReplayRequestEvent(
                        contentPayload);
                processReplayRequest(replayRequest);
                return;
            }
            System.out.println(moduleName + "Unknown exchange type [" +
                               exchangeType
                               + "] received. ");
            return;
        }
        System.out.println(moduleName + "Unknown event type (" + eventType
                           + ") received.");
    }

    /** end onEvent() */
    public void onServiceException(ServiceException serviceException) {
        System.out.println(moduleName + serviceException);
    }
}
